//============================================================================
// Copyright (C) 2009-2013 Brett R. Jones 
//
// Issued to public domain 2013
//
// brettjones1900@gmail.com
// http://www.p2panarchy.com
//============================================================================

#include "VxNetLib.h"
#include <VxUtilLib/VxSktUtil.h>
#include <VxUtilLib/VxParse.h>
#include <VxUtilLib/VxGlobals.h>
#include "VxSktAccept.h"
#include <stdio.h>
#include <memory.h>

//! thread function to listen for connections
static U32 SktAcceptMgrThreadFunc(  void * pvContext );

int VxSktAcceptMgr::m_iAcceptMgrCnt = 0;				// number of accept managers that have been created

//============================================================================
//! thread function to listen for connections
static U32 SktAcceptMgrThreadFunc(  void * pvContext )
{
	VxThread * poThread = (VxThread *)pvContext;
	poThread->SetIsThreadRunning( true );
	VxSktAcceptMgr * poMgr = (VxSktAcceptMgr *)poThread->GetUserParam();

	log_msg( 0, "#### SktAcceptMgr: Mgr id %d Listen port %d thread started\n", poMgr->m_iMgrId, poMgr->m_u16ListenPort );
	poMgr->ListenForConnectionsToAccept( poThread );

	// quitting
	log_msg( 0, "#### SktAcceptMgr: Mgr id %d Listen port %d thread quiting\n", poMgr->m_iMgrId, poMgr->m_u16ListenPort );
	//! Thread calls this just before exit
	poThread->ThreadAboutToExit();
	return 0;
}

//============================================================================
//=== constructor ===//
VxSktAcceptMgr::VxSktAcceptMgr()
	: VxSktBaseMgr()
	, m_u16ListenPort(0)		// what port to listen on
	, m_iActiveListenSktCnt(0)		// listening socket
{
	m_iAcceptMgrCnt++;
	m_iMgrId = m_iAcceptMgrCnt;
	m_eSktMgrType = eSktMgrTypeTcpAccept;
	memset( m_aoListenSkts, 0, sizeof( m_aoListenSkts ) );
}

//============================================================================
//=== destructor ===//
VxSktAcceptMgr::~VxSktAcceptMgr()
{
	stopListening();
}

//============================================================================
// overrides VxSktBaseMgr
void VxSktAcceptMgr::sktMgrShutdown( void )
{
	stopListening();
	VxSktBaseMgr::sktMgrShutdown();
}

//============================================================================
//! make a new socket... give derived classes a chance to override
VxSktBase * VxSktAcceptMgr::makeNewSkt( void )				
{ 
	return new VxSktAccept(); 
}

//============================================================================
//! Startup and start listening
RCODE VxSktAcceptMgr::startListening( U16 u16ListenPort )
{
	log_msg( LOG_INFO, "VxSktAcceptMgr::startListening port %d\n", u16ListenPort );
	RCODE rc = 0;
	m_u16ListenPort = u16ListenPort;

	//
	// By setting the AI_PASSIVE flag in the hints to getaddrinfo, we're
	// indicating that we intend to use the resulting address(es) to bind
	// to a socket(s) for accepting incoming connections.  This means that
	// when the Address parameter is NULL, getaddrinfo will return one
	// entry per allowed protocol family containing the unspecified address
	// for that family.
	//
	char *Address = NULL;

	struct addrinfo *AddrInfo, *AI;
	struct addrinfo oHints;
	VxFillHints( oHints );

	char as8Port[10];
	sprintf( as8Port, "%d", u16ListenPort );

	int RetVal = getaddrinfo(Address, as8Port, &oHints, &AddrInfo);
	if( 0 != RetVal ) 
	{
		log_msg( LOG_INFO, "getaddrinfo failed with error %d: %s\n", RetVal, gai_strerror(RetVal));
		return RetVal;
	}
	//
	// For each address getaddrinfo returned, we create a new socket,
	// bind that address to it, and create a queue to listen on.
	//
	for(AI = AddrInfo; AI != NULL; AI = AI->ai_next) 
	{
		// Highly unlikely, but check anyway.
		if( FD_SETSIZE == m_iActiveListenSktCnt ) 
		{
			log_msg( LOG_ERROR, "getaddrinfo returned more addresses than we could use.\n");
			break;
		}
		// This example only supports PF_INET and PF_INET6.
		if ((AI->ai_family != PF_INET) && (AI->ai_family != PF_INET6))
			continue;

		// Open a socket with the correct address family for this address.
		m_aoListenSkts[m_iActiveListenSktCnt] = socket(AI->ai_family, AI->ai_socktype, AI->ai_protocol);
		if( INVALID_SOCKET == m_aoListenSkts[m_iActiveListenSktCnt] ) 
        {
			log_msg( LOG_ERROR,  "socket() failed with error %d: %s\n",
				VxGetLastError(), DescribeSktErr(VxGetLastError()));
			continue;
		}

#if 0 //#ifdef WIN_PLATFORM
		if ((PF_INET6 == AI->ai_family ) &&
			IN6_IS_ADDR_LINKLOCAL((IN6_ADDR *) INETADDR_ADDRESS(AI->ai_addr)) &&
			(((SOCKADDR_IN6 *) (AI->ai_addr))->sin6_scope_id == 0)
			) 
		{
			log_msg( LOG_ERROR, "IPv6 link local addresses should specify a scope ID!\n");
		}
#endif // WIN_PLATFORM
		//
		// bind() associates a local address and port combination
		// with the socket just created. This is most useful when
		// the application is a server that has a well-known port
		// that clients know about in advance.
		//
		if( SOCKET_ERROR == bind( m_aoListenSkts[ m_iActiveListenSktCnt ], AI->ai_addr, (int) AI->ai_addrlen) ) 
		{
			log_msg( LOG_ERROR, "bind() failed with error %d: %s\n",
				VxGetLastError(), DescribeSktErr(VxGetLastError()));
            VxCloseSkt( m_aoListenSkts[ m_iActiveListenSktCnt ] );
			continue;
		}
		//
		// So far, everything we did was applicable to TCP as well as UDP.
		// However, there are certain fundamental differences between stream
		// protocols such as TCP and datagram protocols such as UDP.
		//
		// Only connection orientated sockets, for example those of type
		// SOCK_STREAM, can listen() for incoming connections.
		//

		if( SOCKET_ERROR == listen( m_aoListenSkts[ m_iActiveListenSktCnt ], 5 ) ) // SOMAXCONN) ) ( SOMAXCONN == maximum number of qued allowed
        {
			log_msg( LOG_ERROR, "listen() failed with error %d: %s\n",
                VxGetLastError(), DescribeSktErr(VxGetLastError()));
            VxCloseSkt( m_aoListenSkts[ m_iActiveListenSktCnt ] );
			m_aoListenSkts[ m_iActiveListenSktCnt ] = INVALID_SOCKET;
			continue;
		}

		log_msg( LOG_INFO, "'Listening' on port %s, protocol %s, protocol family %s\n",
			as8Port, 
			"TCP",
			(AI->ai_family == PF_INET) ? "PF_INET" : "PF_INET6");
		m_iActiveListenSktCnt++;
		break;
	}

	freeaddrinfo(AddrInfo);

	if( 0 == m_iActiveListenSktCnt ) 
	{
		log_msg( LOG_ERROR, "### Fatal error: unable to serve on any address.\n");
		return -1;
	}
	log_msg( LOG_INFO, "VxSktAcceptMgr::startListening starting thread\n" );
	// make a useful thread name
	std::string strThreadName;
	StdStringFormat( strThreadName, "VxSktAcceptMgr%d", m_iMgrId );
	rc = m_oListenThread.StartThread( (VX_THREAD_FUNCTION_T)SktAcceptMgrThreadFunc, this, strThreadName.c_str() );
	VxSleep(200);
	return rc;
}

//============================================================================
//! Shutdown
RCODE VxSktAcceptMgr::stopListening( void )
{
	log_msg( LOG_SKT, "VxSktAcceptMgr: Mgr %d stop listening created ? %d\n", m_iMgrId,  m_oListenThread.IsThreadCreated() );
	// kill previous thread if running
    m_oListenThread.Abort( TRUE );
	for( int i = 0; i < m_iActiveListenSktCnt; i++ )
	{
		if( INVALID_SOCKET != m_aoListenSkts[ i ] )
		{
			log_msg( 0, "VxSktAcceptMgr: Mgr %d closing listen skt %d\n", m_iMgrId, i );
			// closing the thread should release it so it can exit

            VxCloseSkt( m_aoListenSkts[ i ] );

			m_aoListenSkts[ i ]= INVALID_SOCKET;
		}
    }
	m_iActiveListenSktCnt = 0;
	if( m_oListenThread.IsThreadCreated() )
	{
		m_oListenThread.KillThread();
	}
	log_msg( LOG_SKT, "VxSktAcceptMgr: Mgr %d stop listening done\n", m_iMgrId );
	return 0;
}

//============================================================================
RCODE VxSktAcceptMgr::AcceptConnection( VxThread * poThread, SOCKET oListenSkt )
{
	RCODE rc = 0;
	// perform accept
	// setup address
	struct  sockaddr_storage oAcceptAddr;
#ifdef WIN_PLATFORM
	int iAcceptAddrLen = sizeof( oAcceptAddr );
#else //LINUX
	socklen_t iAcceptAddrLen = sizeof( oAcceptAddr );
#endif // LINUX
	memset( &oAcceptAddr, 0, sizeof( struct sockaddr ) );

	// do accept
	SOCKET oAcceptSkt = accept( oListenSkt, (struct sockaddr *)&oAcceptAddr, &iAcceptAddrLen );
#ifdef DEBUG_SKT
	log_msg( LOG_INFO, "SktAcceptMgr: accepting new skt 0x%x\n", oAcceptSkt );
#endif // DEBUG_SKT
	if( poThread->isAborted() || 
		( INVALID_SOCKET == oListenSkt ) )
	{
		// abort
		log_msg( LOG_INFO, "SktAcceptMgr: abort or invalid socket\n" );
		return -1;
	}
	if( poThread->isAborted() || INVALID_SOCKET == oAcceptSkt ) 
	{
		// error
		rc = VxGetLastError();
		log_msg( LOG_INFO, "SktAcceptMgr: invalid socket rc %d\n", rc );
		return -1;
	}

	// valid accept socket
	if( m_aoSkts.size() >= m_u32MaxConnections )
	{
		log_msg( LOG_INFO, "SktAcceptMgr: reached max connections %d\n", m_u32MaxConnections );
		// we have reached max connections
		// just close it immediately
        VxCloseSktNow( oAcceptSkt );

		// sleep awhile
		VxSleep( 200 );
		return 0; // keep running until number of connections clear up
	}

	// add a skt to our list
	
	VxSktAccept * poSkt = (VxSktAccept *)this->makeNewSkt();
	m_oSktMgrMutex.Lock(__FILE__, __LINE__); // dont let other threads mess with array while we add
	m_aoSkts.push_back( poSkt );
	m_oSktMgrMutex.Unlock(__FILE__, __LINE__);

	// do tell skt to do accept stuff
	poSkt->m_oSocket = oAcceptSkt;
	poSkt->SetRecieveCallback( m_pfnOurRecieve );
	poSkt->SetTransmitCallback( m_pfnOurTransmit );
#ifdef DEBUG_SKT
	log_msg( LOG_INFO, "SktAcceptMgr: doing accept\n" );
#endif // DEBUG_SKT
	RCODE rcAccept = poSkt->DoAccept( this, oAcceptAddr );
	if( rcAccept || poThread->isAborted() || INVALID_SOCKET == oListenSkt )
	{
		poSkt->CloseSkt(67823);
		log_msg( LOG_ERROR, "SktAcceptMgr: error %d doing accept\n", rc );
		// if error occurred then remove it from our list or it
		// will never get removed
		m_oSktMgrMutex.Lock(__FILE__, __LINE__); // dont let other threads mess with array while we remove
		m_aoSkts.pop_back();
		m_oSktMgrMutex.Unlock(__FILE__, __LINE__);
		delete poSkt;
		return -1;
	}
	return 0;
}

//============================================================================
void VxSktAcceptMgr::ListenForConnectionsToAccept( VxThread * poThread )
{
	int iSelectResult;
	RCODE rc = 0;
	log_msg( LOG_INFO, "Listen Thread start\n" );
	// use a timeout so that thread can exit if aborted
	fd_set					oSocketFileDescriptors;
	struct timeval			oListenTimeout;

	log_msg( LOG_INFO, "Listen Thread setting descriptors\n" );
	FD_ZERO( &oSocketFileDescriptors );
    bool bContinueListen = true;
    while( ( false == poThread->isAborted() ) &&
           ( false == VxIsAppShuttingDown() ) &&
           bContinueListen )
	{
		// Check to see if we have any sockets remaining to be served
		// from previous time through this loop.  If not, call select()
		// to wait for a connection request or a datagram to arrive.
		//log_msg( LOG_INFO, "Listen Thread checking descriptors\n" );
        int iSelectedIdx = -1;
        for( int iSktSetIdx = 0; iSktSetIdx < m_iActiveListenSktCnt; iSktSetIdx++ )
		{
            if( FD_ISSET( m_aoListenSkts[ iSktSetIdx ], &oSocketFileDescriptors) )
			{
                iSelectedIdx = iSktSetIdx;
				break;
			}
		}

        if( poThread->isAborted() ||
            VxIsAppShuttingDown() )
		{
			break;
        }

        if( 0 > iSelectedIdx )
		{
			// none selected.. do select
            for( int iSktSetIdx = 0; iSktSetIdx < m_iActiveListenSktCnt; iSktSetIdx++ )
			{
                //log_msg( LOG_INFO, "Listen Thread setting descriptor %d\n", iSktSetIdx );
                FD_SET( m_aoListenSkts[ iSktSetIdx ], &oSocketFileDescriptors);
                oListenTimeout.tv_sec	= 3;
                oListenTimeout.tv_usec	= 0;
				iSelectResult = select( m_iActiveListenSktCnt, &oSocketFileDescriptors, 0, 0, &oListenTimeout );
				//log_msg( LOG_INFO, "Listen Thread select result %d\n", iSelectResult );
				// timed out.. check if we need to abort
				if( poThread->isAborted() ) 
				{
					log_msg( LOG_INFO, "Listen Thread Exiting\n" );
                    bContinueListen = false;
					// exit listen loop
					break;
				}

				if( 0 == iSelectResult )
				{
                    break;
				}
				else if( ( 0 < iSelectResult ) && ( iSelectResult <= m_iActiveListenSktCnt ) )
				{
					// at least one valid selected
					break;
				}
				else if( (4 == errno) || poThread->isAborted() )
				{
					// errno 4 can occur if in linux and app gets CTRL-C signal
                    bContinueListen = false;
					break;
				}
				else
				{
					// ignore the error and try again
					VxSleep(1);
					continue;
				}
			}
		}
        if( poThread->isAborted() ||
            VxIsAppShuttingDown() ||
            ( false == bContinueListen ) )
		{
			break;
		}
		for( iSelectedIdx = 0; iSelectedIdx < m_iActiveListenSktCnt; iSelectedIdx++ ) 
		{
			if( FD_ISSET( m_aoListenSkts[iSelectedIdx], &oSocketFileDescriptors) ) 
			{
#ifdef DEBUG_SKT
                log_msg( LOG_INFO, "#### SktAcceptMgr::AcceptConnection: accepting at index %d\n", iSelectedIdx );
#endif // DEBUG_SKT
				FD_CLR( m_aoListenSkts[iSelectedIdx], &oSocketFileDescriptors);

				rc = AcceptConnection( poThread, m_aoListenSkts[iSelectedIdx] );
#ifdef DEBUG_SKT
				if( rc )
				{
					log_msg( LOG_ERROR, "#### SktAcceptMgr::AcceptConnection: error %d %s doing accept\n", rc, DescribeSktErr(rc) );
				}
				else
				{
					log_msg( LOG_INFO, "#### SktAcceptMgr::AcceptConnection: success doing accept\n" );
				}
#endif // DEBUG_SKT
				break;
			}
		}
	}
}
